package com.atguigu.spark.util

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.dstream.ReceiverInputDStream

object SparkStreaming07_Window {
  def main(args: Array[String]): Unit = {

    // TODO Spark Streaming Resume
    val ssc = StreamingContext.getOrCreate(
      "cp",
      ()=>{
        val conf = new SparkConf().setMaster("local[*]").setAppName("WordCount").set("spark.testing.memory", "2147480000")
        val ssc = new StreamingContext(conf, Seconds(3))
        ssc.checkpoint("cp")

        // socket流接收时是一行字符串接收
        val socketDS: ReceiverInputDStream[String] = ssc.socketTextStream("localhost", 9999)

        val wordDS = socketDS.flatMap(_.split(" "))

        val wordToOneDS = wordDS.map((_, 1))

        //val wordToCountDS = wordToOneDS.reduceByKey(_+_)
        // TODO updateStateByKey方法可以传递一个参数
        // 这个参数的类型为函数类型：（Seq[Int], Option[S]） => Option[S]
        // The checkpoint directory has not been set. Please set it by StreamingContext.checkpoint().
        wordToOneDS.updateStateByKey(
          (seq, buff:Option[Int]) => {
            val currentTotal: Int = seq.sum
            val oldCnt: Int = buff.getOrElse(0)
            Option(currentTotal + oldCnt)
          }
        ).print()
        ssc
      }
    )
    ssc.start()
    ssc.awaitTermination()
  }
}
